package cn.cdqf.rabbitmq.rabbit;

import cn.cdqf.rabbitmq.util.RedisUtils;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.context.annotation.Bean;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.util.HashMap;
import java.util.Map;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;

@Component
@Slf4j
public class RabbitSender {
    //springboot自动注入的 rabbitmq的实例
    @Autowired
    private RabbitTemplate rabbitTemplate;
    //定义confirm 确认监听回调函数  是否成功将消息 发送到rabbitmq的broker 都会执行该方法
    final RabbitTemplate.ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {
        //先看第二个参数 就是消息是否发送到了broker的结果
        //第一个参数为传递的消息的信息 包含了全局唯一id
        @Override
        public void confirm(CorrelationData correlationData, boolean flag, String s) {
            log.info("发送到broker的结果为：{}",flag);
            log.info("消息的id为：{}",correlationData.getId());
            //说明没有成功发送到交换机 ，就可以做一定业务逻辑处理
            if(!flag){
                log.info("没有成功发送到交换机，做对应的处理.......");
            }
        }
    };
    //定义return的监听回调函数  不能将消息路由到了指定queue 就会指定该回调方法
    final RabbitTemplate.ReturnCallback returnCallback = new RabbitTemplate.ReturnCallback() {
        /**
         * @param message   发送的消息
         * @param relayCode 错误码
         * @param relayText 错误文本信息
         * @param exchange  发送的交换机
         * @param routingKey 发送的路由键
         */
        @Override
        public void returnedMessage(org.springframework.amqp.core.Message message, int relayCode, String relayText,
                                    String exchange, String routingKey) {
            log.info("该失败消息发送到的交换机为：{}",exchange);
            log.info("该失败消息的路由键为：{}",routingKey);
            log.info("该失败消息的错误码为：{}",relayCode);
            log.info("该失败消息的错误文本信息为：{}",relayText);
            //重发消息
            Message message1 = (org.springframework.messaging.Message)(ByteToObject(message.getBody()));
            String messageId = (String) message1.getHeaders().get("messageId");
            //分布式锁  第三个参数是 正常走业务的一个时间+5
           if(redisUtils.hsetnx("rabbit_"+messageId,"sendLock",System.currentTimeMillis()+2000*5)) {
              try {
                  //多线程完成    如果断电  或者服务器崩溃
                 CompletableFuture<Void> completableFuture = CompletableFuture.runAsync(()->{
                      sendMessageById("rabbit_"+messageId);
                  },getAsyncExecutor);
                  completableFuture.join();
              }finally {

                  //释放锁
                  redisUtils.hdel("rabbit_"+messageId,"sendLock");
              }
           }else {
               Object object = redisUtils.hget("rabbit_"+messageId,"sendLock");
               //避免死锁,只能补偿  极端情况
               if(object!=null&&Long.valueOf(String.valueOf(object))<System.currentTimeMillis()){
                   //在把时间改变一次
                   redisUtils.hset("rabbit_"+messageId,"sendLock",System.currentTimeMillis()+2000*5);
                   //发送次数减一  多发一次而已
                   redisUtils.hincr("rabbit_"+messageId,"sendCount",-1);
                   try {
                       //多线程完成
                       CompletableFuture<Void> completableFuture = CompletableFuture.runAsync(()->{
                           sendMessageById("rabbit_"+messageId);
                       },getAsyncExecutor);
                       completableFuture.join();
                   }finally {
                       //释放锁
                       redisUtils.hdel("rabbit_"+messageId,"sendLock");
                   }
               }
           }
        }
    };
    @Async
    public void sendMessageById(String messageKey){
        Map<Object, Object> map =  redisUtils.hmget(messageKey);
        if(map==null){//消费成功
            return;
        }
        //判断次数  消费状态
        Integer consumerStatus = Integer.valueOf(String.valueOf(map.get("consumerStatus"))) ;
        //发送次数
        Integer sendCount = Integer.valueOf(String.valueOf(map.get("sendCount"))) ;
        long sendTimeOut = Long.valueOf(String.valueOf(map.get("sendTimeOut"))) ;

        //重发次数
        if(sendCount<3){
        //未消费
        if(consumerStatus!=1){
            //已经过期
            if(sendTimeOut<= System.currentTimeMillis()){
                String message = (String) map.get("message");
                String exchange = (String) map.get("exchange");
                String routKey = (String) map.get("routKey");
               String properties = (String) map.get("properties");
                try {
                   Map hashMap =  objectMapper.readValue(properties,HashMap.class);
                  send(message,hashMap,exchange,routKey,++sendCount);
                } catch (IOException e) {
                    e.printStackTrace();
                }

            }
        }
        }else{

        }
    }
    //丢到死信队列
   /* @Bean("deadQueue")
    public Queue deadQueue(){
        return QueueBuilder.durable().autoDelete().exclusive().build();
    }
    @Bean("deadExchange")
    public Exchange deadExchange(){
        return ExchangeBuilder.topicExchange("topic").build();
    }
    @Bean
    public Binding binding(){
        Binding dead = BindingBuilder.bind(deadQueue()).to(deadExchange()).with("dead").noargs();
        return dead;
    }
*/
    /**
     * byte转对象
     * @param bytes
     * @return
     */
    private Object ByteToObject(byte[] bytes) {
        Object obj = null;
        try {
            // bytearray to object
            ByteArrayInputStream bi = new ByteArrayInputStream(bytes);
            ObjectInputStream oi = new ObjectInputStream(bi);

            obj = oi.readObject();
            bi.close();
            oi.close();
        } catch (Exception e) {
            System.out.println("translation" + e.getMessage());
            e.printStackTrace();
        }
        return obj;
    }
    /**
     *
     * @param message  发送的消息
     * @param properties  消息的附加属性
     */
    @Autowired
    private Executor getAsyncExecutor;
    @Autowired
    private RedisUtils redisUtils;
    @Autowired
    private ObjectMapper objectMapper;
    public void send(String message, Map properties,String exchange,String routKey){
        if(properties==null){
            System.out.println("map为null");
            properties = new HashMap();
        }
        //自己生成一个消息ID作为全局标识 rabbit123456 只是测试用
        properties.put("messageId","rabbit123456");
        send(message,properties,exchange,routKey,1);
    }
    @Async
    public void send(String message, Map properties,String exchange,String routKey,int count){

        MessageHeaders messageHeaders = new MessageHeaders(properties);
        //组合了 message 消息体与properties消息属性的消息
        Message<String> msg = MessageBuilder.createMessage(message, messageHeaders);
        //加入消息确认模式的监听
        rabbitTemplate.setConfirmCallback(confirmCallback);
        //加入消息return模式的监听
        rabbitTemplate.setReturnCallback(returnCallback);
        //1.交换机名称  2.routingKey 3.发送的消息数据 4.该消息的信息 例如全局唯一的id设置
        //这儿全局唯一ID先使用123456
        CorrelationData correlationData = new CorrelationData(String.valueOf(properties.get("messageId")));
        if(count==1){
            //java是伪泛型  发送成功才存入redis
            Map maps = new HashMap();
            try {
                maps.put("properties",objectMapper.writeValueAsString(properties));
            } catch (JsonProcessingException e) {
                e.printStackTrace();
            }
            maps.put("message",message);
            //发送了几次
            maps.put("sendCount",1+"");
            //状态
            maps.put("sendStatus","0");
            maps.put("sendTimeOut",(System.currentTimeMillis()+20*1000)+"");
            maps.put("consumerStatus","0");
            maps.put("exchange",exchange);
            maps.put("routKey",routKey);
            redisUtils.hmset("rabbit_"+properties.get("messageId"),maps);
        }else{
            //不是第一次发送
            redisUtils.hset("rabbit_"+properties.get("messageId"),"sendCount",count+"");
            redisUtils.hset("rabbit_"+properties.get("messageId"),"sendTimeOut",(System.currentTimeMillis()+20*1000)+"");
        }
       /* MessagePostProcessor messagePostProcessor =  message1->{
            //设置延迟队列：双写一致性，多线程延迟2秒删除
            //顺序消费：只会在特定业务才有，我们的项目没有这个业务，
            // 队列定义的有个参数是独占，就会保证消息被一个消费全部抢到
            //basicQos(1000)
            //消息id我自己设计，生成消息的时候让id有序
            //消息可靠性  幂等性 千万级消息  限流  百万条消息 有几条消息丢失  basicNack（false）补偿处理
            //流程
            //excel:java可以读取 poi  ，系统能不能承受
            message1.getMessageProperties().setDelay(1000);
            return message1;
        };
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setDelay(1000);*/


        /*MessagePostProcessor messagePostProcessor =message1->{
            MessageProperties messageProperties = message1.getMessageProperties();
            //毫秒 延迟
            messageProperties.setDelay(2000);
            //过期时间 毫秒
            messageProperties.setExpiration("2000");
            //消息的编号 顺序消费
            messageProperties.setDeliveryTag(redisUtils.incrBy("deliveryTag",1));
            return message1;
        };
*/
       // rabbitTemplate.convertAndSend(exchange,routKey,msg,messagePostProcessor,correlationData);
       rabbitTemplate.convertAndSend(exchange,routKey,msg,correlationData);


    }
}

